import asyncio
import os
import queue
import threading
from copy import copy, deepcopy
from queue import Queue

from DrissionPage._base.chromium import Chromium
from DrissionPage._configs.chromium_options import ChromiumOptions
from concurrent.futures import ThreadPoolExecutor, as_completed, wait

from DrissionPage._functions.settings import Settings
from DrissionPage.errors import ElementNotFoundError, WaitTimeoutError, PageDisconnectedError
from fake_useragent import UserAgent
from sqlalchemy import func
from sqlalchemy.exc import OperationalError
from sqlmodel import select
from urllib3.exceptions import InvalidChunkLength

from config.db import get_session
from config.get_db import get_db
from constant.crawler_constant import HouseDetailKey, JiNanUrlId, CrawlerId
from crawler_handler.abstract_crawler_handler import AbstractCrawlerHandler
from decorators.exception_decorator import crawler_exception_handler, async_crawler_exception_handler
from dict.crwaler_dict import house_state_dict, project_base_info_dict
from entity.model.community_model import CommunityDataCollection
from entity.model.crawler_record_model import CommunityProjectRecord, CommunityBuildingRecord
from entity.model.jinan.crawler_record_model import JiNanDataCollectionCommunity, JiNanDataCollectionBuilding, \
    JiNanDataCollectionExtra, JiNanDataCollectionUnit, JiNanDataCollectionRoom
from entity.schema.crawler_schema import CrawlerJiNanArgs, CrawlerCommunityByName
from enums.crawler_enum import ProjectType, HouseState
from util.crawler_util import get_query_params, generate_uuid
from util.data_change_util import change_to_float, change_to_int, get_date_str, get_digit
from util.mylog import my_logger

class JinanHandler(AbstractCrawlerHandler):
    executor = ThreadPoolExecutor(max_workers=(os.cpu_count() or 1) * 4, thread_name_prefix='com_pro')
    building_executor = ThreadPoolExecutor(max_workers=(os.cpu_count() or 1) * 4, thread_name_prefix='building')
    room_executor = ThreadPoolExecutor(max_workers=(os.cpu_count() or 1) * 4, thread_name_prefix='room')
    data_executor = ThreadPoolExecutor(max_workers=(os.cpu_count() or 1) * 4, thread_name_prefix='data_detail')
    community_executor = ThreadPoolExecutor(max_workers=(os.cpu_count() or 1) * 4, thread_name_prefix='community')

    community_url = 'https://jncc.jinan.gov.cn/jnfdcinfo/jnfdcweb/onsaling/index_{}.shtml?zn={}&pu={}&pn={}&en={}'
    community_project_url_prefix = 'https://jncc.jinan.gov.cn/jnfdcinfo/jnfdcweb/onsaling/show_{}.shtml?prjno={}'
    build_table_url_prefix = 'https://jncc.jinan.gov.cn/jnfdcinfo/jnfdcweb/onsaling/viewhouse.shtml?fmid={}'
    build_info_url_prefix = 'https://jncc.jinan.gov.cn/jnfdcinfo/jnfdcweb/onsaling/bshow.shtml?bno={}'
    room_link_prefix = 'https://jncc.jinan.gov.cn/jnfdcinfo/jnfdcweb/onsaling/viewDiv.shtml?fid={}'
    zones_dict = {
        370102: "历下区",
        370103: "市中区",
        370105: "天桥区",
        370112: "历城区",
        370113: "长清区",
        370104: "槐荫区",
        370114: "章丘区",
        370115: "济阳区",
        370116: "莱芜区",
        370117: "钢城区",
        370124: "平阴县",
        370126: "商河县",
        370171: "高新区",
        370192: "先行区",
        371203: "莱芜高新区"
    }

    def __init__(self, browser: Chromium, crawler_args: CrawlerJiNanArgs):
        super().__init__(browser)
        self.crawler_args = crawler_args
        self.community_queue_task = Queue()
        self.building_queue_task = Queue()
        self.room_detail_queue_task = Queue()


    @crawler_exception_handler
    def get_community_building_link(self, page_num: int, community_id):
        community_link = JinanHandler.community_project_url_prefix.format(page_num, community_id)
        with JinanHandler.fetch_page(community_link) as page:
            community_building_list = page.s_eles('.list-main')
        if not community_building_list:
            my_logger.warning(f"没有楼盘信息: {community_id}")
            return False, community_id
        need_crawler_community_building_model_list = []
        for item in community_building_list:
            # 楼盘表链接
            community_table = item.ele('#houseshow')
            if not community_table or not community_table.link:
                my_logger.warning(f"没有楼栋表链接: {community_id}:{item}")
                continue
            community_building_model = JiNanDataCollectionBuilding()
            community_building_model.community_id = community_id
            # 楼栋id同楼盘表id
            community_building_id = JinanHandler.get_community_table_id(community_table.link)
            community_building_model.collect_building_id = community_building_id
            # 获取楼盘基本信息
            is_success, res = self.get_building_info(community_building_model)
            if is_success:
                need_crawler_community_building_model_list.append(res)
        # my_logger.debug(len(need_crawler_community_building_model_list))
        return True, need_crawler_community_building_model_list

    def collect_community_building(self, all_page: int, community_id: str):
        for page_num in range(1, all_page + 1):
            my_logger.debug(f"正在获取第{page_num}页")
            self.building_queue_task.put((page_num, community_id))

    # 获取楼栋信息
    @crawler_exception_handler
    def get_building_info(self, community_building_model: JiNanDataCollectionBuilding):
        # 获取楼盘基本信息
        with self.fetch_page(JinanHandler.build_info_url_prefix.format(community_building_model.collect_building_id)) as page:
            building_info_table = page.s_ele('.message_table')
            tr_list = building_info_table.eles('@tag()=tr')
            # 楼栋名称
            tr_build_name = tr_list[1].ele('#lblFMonoName').text
            # 许可证号
            tr_certificate_no = tr_list[-1].eles('.massage_bg')[1].text
            # 监管银行和监管账号
            band_account = tr_list[-2].eles('.massage_bg')
            tr_bank = band_account[1].text
            tr_account = band_account[-1].text
            community_building_model.building_number = tr_build_name
            community_building_model.certificate_no = tr_certificate_no
            if tr_bank is not None and not "特此说明" in tr_bank:
                community_building_model.regulatory_banks = tr_bank
            community_building_model.regulatory_account = tr_account
            # 是否是住宅
            form_list_main_list = page.eles('.list-main')
            if form_list_main_list:
                is_house_flag = False
                for form_list_main in form_list_main_list:
                    form_ul = form_list_main.child()
                    li_text_list = form_ul.s_eles('.ellipsis')
                    if li_text_list[0].text == '住宅':
                        is_house_flag = True
                        extra_model = JiNanDataCollectionExtra()
                        extra_model.building_id = community_building_model.collect_building_id
                        extra_model.can_sale_count = change_to_int(li_text_list[-2].text)
                        with get_session() as session:
                            stat = select(JiNanDataCollectionExtra.can_sale_count).where(
                                JiNanDataCollectionExtra.building_id == community_building_model.collect_building_id)
                            extra_record = session.exec(stat).first()
                            if extra_record is not None and extra_record == extra_model.can_sale_count:
                                self.bulk_save_or_update(JiNanDataCollectionBuilding, [community_building_model],[CrawlerId.BUILDING_ID])
                                return False, community_building_model
                            else:
                                self.bulk_save_or_update(JiNanDataCollectionExtra, [extra_model],[CrawlerId.BUILDING_ID])
                        break
                # 只要住宅
                if not is_house_flag:
                    return False, community_building_model
            else:
                # 啥房都没有
                extra_model = JiNanDataCollectionExtra()
                extra_model.building_id = community_building_model.collect_building_id
                extra_model.can_sale_count = 0
                self.bulk_save_or_update(JiNanDataCollectionBuilding, [community_building_model],
                                         [CrawlerId.BUILDING_ID])
                self.bulk_save_or_update(JiNanDataCollectionExtra, [extra_model],[CrawlerId.BUILDING_ID])
                return False, community_building_model
        return True, community_building_model

    @async_crawler_exception_handler
    async def async_get_building_info(self, community_building_model: JiNanDataCollectionBuilding):
        # 获取楼盘基本信息
        async with self.async_fetch_page(JinanHandler.build_info_url_prefix.format(community_building_model.collect_building_id)) as page:
            building_info_table = page.s_ele('.message_table')
            tr_list = building_info_table.eles('@tag()=tr')
            # 楼栋名称
            tr_build_name = tr_list[1].ele('#lblFMonoName').text
            # 许可证号
            tr_certificate_no = tr_list[-1].eles('.massage_bg')[1].text
            # 监管银行和监管账号
            band_account = tr_list[-2].eles('.massage_bg')
            tr_bank = band_account[1].text
            tr_account = band_account[-1].text
            community_building_model.building_number = tr_build_name
            community_building_model.certificate_no = tr_certificate_no
            if tr_bank is not None and not "特此说明" in tr_bank:
                community_building_model.regulatory_banks = tr_bank
            community_building_model.regulatory_account = tr_account
            # 是否是住宅
            form_list_main_list = page.eles('.list-main')
            if form_list_main_list:
                is_house_flag = False
                for form_list_main in form_list_main_list:
                    form_ul = form_list_main.child()
                    li_text_list = form_ul.s_eles('.ellipsis')
                    if li_text_list[0].text == '住宅':
                        is_house_flag = True
                        extra_model = JiNanDataCollectionExtra()
                        extra_model.building_id = community_building_model.collect_building_id
                        extra_model.can_sale_count = change_to_int(li_text_list[-2].text)
                        async for session in get_db():
                            stat = select(JiNanDataCollectionExtra.can_sale_count).where(
                                JiNanDataCollectionExtra.building_id == community_building_model.collect_building_id)
                            extra_record = await session.exec(stat)
                            extra_record = extra_record.first()
                            if extra_record:
                                if extra_record == change_to_int(li_text_list[-2].text):
                                    await self.async_bulk_save_or_update(JiNanDataCollectionBuilding, [community_building_model],[CrawlerId.BUILDING_ID])
                                    return False, community_building_model
                        await self.async_bulk_save_or_update(JiNanDataCollectionExtra, [extra_model],
                                                                       [CrawlerId.BUILDING_ID])
                        break
                if not is_house_flag:
                    return False, community_building_model
            else:
                # 啥房都没有
                extra_model = JiNanDataCollectionExtra()
                extra_model.building_id = community_building_model.collect_building_id
                extra_model.can_sale_count = 0
                await self.async_bulk_save_or_update(JiNanDataCollectionExtra, [extra_model],
                                                     [CrawlerId.BUILDING_ID])
        return True, community_building_model

    @crawler_exception_handler
    def get_community_building_info(self, community_id: str):
        community_link = JinanHandler.community_project_url_prefix.format(1, community_id)
        with self.fetch_page(community_link) as community_building_page:
            # 可能没楼栋
            all_counts, all_page = JinanHandler.get_all_page_count(community_building_page)
            if all_counts is None or all_page is None:
                my_logger.warning(f"获取楼栋数量失败: {community_id}")
                return False, community_id
            # 一个项目中每行的楼栋
            self.collect_community_building(all_page, community_id)


    @async_crawler_exception_handler
    async def async_get_community_building_info(self, community_id: str):
        model = JiNanDataCollectionBuilding()
        model.community_id = community_id
        community_link = JinanHandler.community_project_url_prefix.format(1, model.community_id)
        async with self.async_fetch_page(community_link) as community_building_page:
            # 可能没楼栋
            all_counts, all_page = await asyncio.to_thread(JinanHandler.get_all_page_count, community_building_page)
            if all_counts is None or all_page is None:
                my_logger.error(f"获取楼栋数量失败: {community_id}")
                return False, community_id
            # 一个项目中每行的楼栋
            community_building_list, fail_community_id_list = await asyncio.to_thread(self.collect_community_building, all_page, model.community_id)
            if fail_community_id_list:
                my_logger.warning(f"获取楼栋信息失败:{fail_community_id_list}")
        need_crawler_community_building_model_list = []
        for item in community_building_list:
            # 楼盘表链接
            community_table = item.ele('#houseshow')
            if not community_table or not community_table.link:
                continue
            community_table_link = community_table.link
            # Shallow copy
            community_building_model = copy(model)
            # 楼栋id同楼盘表id
            community_building_id = JinanHandler.get_community_table_id(community_table_link)
            community_building_model.collect_building_id = community_building_id
            # 获取楼盘基本信息
            is_success, res = await self.async_get_building_info(community_building_model)
            if is_success:
                need_crawler_community_building_model_list.append(res)
            else:
                # todo 失败如何处理
                my_logger.warning(f"获取楼栋信息失败:{community_building_id}")
        JinanHandler.bulk_save_or_update(JiNanDataCollectionBuilding, need_crawler_community_building_model_list, [CrawlerId.BUILDING_ID])
        return True, need_crawler_community_building_model_list

    @crawler_exception_handler
    def get_community_link(self, page_num, county_id, county_name):
        link = JinanHandler.community_url.format(page_num, county_id, self.crawler_args.project_type.value, self.crawler_args.project_name, '')
        with self.fetch_page(link) as page:
            form_list = page.ele('.form-list')
            if not form_list:
                my_logger.warning(f"获取链接失败:{link}")
                return False, []
            community_model_list = []
            community_item_list = form_list.s_eles('.list-main')
            for community_item in community_item_list:
                li_list = community_item.child().children()
                community_link = li_list[1].ele('t:a').link
                if not community_link:
                    my_logger.error(f"获取链接失败:{community_item}")
                community_id = JinanHandler.get_community_link_id(community_link)
                community_model = JiNanDataCollectionCommunity()
                community_model.community_name = li_list[1].child().attr('title')
                community_model.address = li_list[2].child().attr('title')
                community_model.developer = li_list[-2].child().attr('title')
                community_model.collect_community_id = community_id
                community_model.province = "山东省"
                community_model.province_id = 37
                community_model.area = "济南市"
                community_model.area_id = 3701
                community_model.county_id = county_id
                community_model.county = county_name
                community_model.data_source = 1
                community_model_list.append(community_model)
        return True, community_model_list

    async def async_get_community_link(self, page_num, county_id, county_name):
        link = JinanHandler.community_url.format(page_num, county_id, self.crawler_args.project_type.value, '', '')
        async with self.async_fetch_page(link) as page:
            form_list = page.ele('.form-list')
            if not form_list:
                my_logger.warning(f"表单为空:{link}")
                return False, []
            community_model_list = []
            community_item_list = form_list.s_eles('.list-main')
            for community_item in community_item_list:
                li_list = community_item.child().children()
                community_link = li_list[1].ele('t:a').link
                if not community_link:
                    my_logger.error(f"获取链接失败:{community_item}")
                    continue
                community_id = JinanHandler.get_community_link_id(community_link)
                community_model = JiNanDataCollectionCommunity()
                community_model.community_name = li_list[1].child().attr('title')
                community_model.address = li_list[2].child().attr('title')
                community_model.developer = li_list[-2].child().attr('title')
                community_model.collect_community_id = community_id
                community_model.province = "山东省"
                community_model.province_id = 37
                community_model.area = "济南市"
                community_model.area_id = 3701
                community_model.county_id = county_id
                community_model.county = county_name
                community_model.data_source = 1
                community_model_list.append(community_model)
        return True, community_model_list

    def collect_community_project(self, all_page: int, county_id, county_name):
        for page_num in range(1, all_page + 1):
            my_logger.debug(f"正在获取第{page_num}页")
            self.community_queue_task.put((page_num, county_id, county_name))

    async def async_collect_community_project(self, all_page: int, county_id, county_name):
        futures = []
        for page_num in range(1, all_page + 1):
            my_logger.debug(f"正在获取第{page_num}页")
            futures.append(self.async_get_community_link(page_num, county_id, county_name))
        return await JinanHandler.process_async_futures(futures)

    # 获取楼盘表里住宅的信息
    def collect_house_info(self, model: JiNanDataCollectionRoom):
        i = 0
        while True:
            try:
                link = JinanHandler.room_link_prefix.format(model.collect_room_id)
                my_logger.debug(f"爬取楼盘表信息:{link}")
                # data = self._create_session().get(link, timeout=40).json()
                with JinanHandler.fetch_data(link) as data:
                # my_logger.info(type(data))
                # my_logger.debug(data)
                    if data == '':
                        my_logger.warning(f"获取房间详细信息失败重试: {link}")
                        return False, model
                    model.building_area = change_to_float(data.get(HouseDetailKey.BUILDING_AREA, None))
                    model.inner_area = change_to_float(data.get(HouseDetailKey.INNER_AREA, None))
                    model.public_area = change_to_float(data.get(HouseDetailKey.PUBLIC_AREA, None))
                    model.purpose = data.get(HouseDetailKey.PURPOSE, None)
                    house_state = data.get(HouseDetailKey.SALE_STATUS, None)
                    if house_state:
                        house_state = change_to_int(house_state)
                        model.sale_status = house_state_dict.get(HouseState._value2member_map_[house_state], "可售")
                    # 备案时间
                    synchrotime = data.get(HouseDetailKey.FILING_TIME, None)
                    if synchrotime:
                        model.filing_time = get_date_str(synchrotime)
                return True, [model]
            except InvalidChunkLength:
                my_logger.warning(f"获取房间详细信息失败重试")
                i += 1
                if i > 3:
                    my_logger.error(f"获取房间详细信息失败: {model}")
                    return False, model
            except Exception as e:
                my_logger.error(f"Error get_house_info task: {e}\n info: {model}")
                return False, model
        # my_logger.warning("dddd")
        # return data

    # 获取楼盘表和房间基本信息(floor, room_number, unit)
    def collect_community_table_and_room_info(self, community_building_model: JiNanDataCollectionBuilding):
        lnk = JinanHandler.build_table_url_prefix.format(community_building_model.collect_building_id)
        i = 0
        while True:
            try:
                with self.fetch_tab(lnk) as tab:
                    body = tab.s_ele('@tag()=body', timeout=15)
                    if not body:
                        my_logger.warning("页面为空!")
                        self.room_detail_queue_task.put((False, community_building_model))
                        return False, community_building_model
                    # my_logger.debug(body.child().text)
                    # return False, community_building_model
                    data_table = body.ele('#floorTable', timeout=30)
                    tab.wait(1)
                    tr_list = data_table.s_eles('@tag()=tr')
                    # 单元映射表
                    unit_map = {}
                    # 单元访问映射表
                    unit_visit_map = {}
                    floor = None
                    max_floor = None
                    room_model_list = []
                    # 房间楼层映射表
                    room_floor_map = {}
                    for idx_row, tr in enumerate(tr_list):
                        td_list = tr.children()
                        for idx_td, td in enumerate(td_list):
                            # 单元
                            if idx_row == 0 and idx_td > 0:
                                unit_model = JiNanDataCollectionUnit()
                                # unit_model.unit_number = JinanHandler.convert_unit(td.text)
                                unit_model.generate_unit_id = generate_uuid()
                                unit_model.unit_number = td.text
                                unit_model.room_number_start = "01"
                                colspan = td.attr('colspan')
                                max_room_number = JinanHandler.convert_room_number(colspan)
                                if not max_room_number:
                                    my_logger.error(f"max_room_number转换失败: {colspan}")
                                    continue
                                unit_model.room_number_end = JinanHandler.convert_room(colspan)
                                unit_model.community_id = community_building_model.community_id
                                unit_model.building_id = community_building_model.collect_building_id
                                unit_visit_map[id(unit_model)] = False
                                # 当前单元所占房号
                                l = len(unit_map)
                                for idx in range(max_room_number):
                                    unit_map[idx + l] = unit_model
                            # 室号
                            # 户
                            if idx_row >= 2:
                                # 楼层
                                if idx_td == 0:
                                    floor = JinanHandler.convert_floor(td.text)
                                    # 最大楼层
                                    if idx_row == 2:
                                        max_floor = floor
                                # id
                                if idx_td > 0:
                                    # 房间号
                                    # room_model = copy(unit_map[idx_td - 1])
                                    room_data_id = td.attr('id')
                                    if room_data_id:
                                        cur_unit = unit_map[idx_td - 1]
                                        cur_unit.total_floor = max_floor
                                        room_model = JiNanDataCollectionRoom()
                                        room_model.collect_room_id = room_data_id
                                        room_model.room_number = td.text
                                        room_model.community_id = community_building_model.community_id
                                        room_model.building_id = community_building_model.collect_building_id
                                        if not unit_visit_map[id(cur_unit)]:
                                            unit_visit_map[id(cur_unit)] = True
                                            with get_session() as session:
                                                stat = select(JiNanDataCollectionRoom.unit_id).where(JiNanDataCollectionRoom.collect_room_id == room_data_id)
                                                unit_id_record = session.exec(stat).first()
                                                # 是否有记录
                                                if unit_id_record:
                                                    cur_unit.generate_unit_id = unit_id_record
                                        room_model.unit_id = cur_unit.generate_unit_id
                                        # 判断房间是否掉下来
                                        current_room_id_value = get_digit(room_data_id)
                                        if current_room_id_value in room_floor_map:
                                            room_model.floor = room_floor_map[current_room_id_value]
                                        else:
                                            room_model.floor = floor
                                        room_model_list.append(room_model)
                                        # 加入表中
                                        room_floor_map[current_room_id_value] = room_model.floor

                                        # my_logger.debug(room_model)
                                    else:
                                        if not room_model_list:
                                            continue
                                        # 拿到前面一个房间的数据(Python 3.7 字典有序)
                                        front_room_id_value = list(room_floor_map.keys())[-1]
                                        # 当前位置的id应是前一个房间的id的值 + 1
                                        cur_room_id_value = front_room_id_value + 1
                                        # 也把当前没有房间的id加入表中
                                        room_floor_map[cur_room_id_value] = floor
                    # upsert 单元信息
                    filtered_units = [unit for unit in unit_map.values() if
                                                                unit.total_floor is not None]
                    self.bulk_save_or_update(JiNanDataCollectionUnit, filtered_units, [CrawlerId.UNIT_ID])
                if i > 0:
                    my_logger.debug(f"重试成功: {community_building_model}")
                self.room_detail_queue_task.put((True, room_model_list))
                return True, room_model_list
            except ElementNotFoundError as e:
                t = 10 * (i + 1)
                my_logger.warning(f"{e}\n 可能是被限制了, 等待{t}秒后重新请求:{i}: link: {lnk}")
                tab.wait(t)
                i += 1
                if i > 3:
                    self.room_detail_queue_task.put((False, community_building_model))
                    return False, community_building_model
            except PageDisconnectedError as e:
                my_logger.warning(f"{e}\n 可能网不好, 等待3秒后重新请求:{i}: link: {lnk}")
                tab.wait(3)
                i += 1
                if i > 3:
                    self.room_detail_queue_task.put((False, community_building_model))
                    return False, community_building_model
            except WaitTimeoutError as e:
                my_logger.warning(e)
                my_logger.warning(f"失败 link:{i}: {lnk}")
                # my_logger.warning("重试")
                tab.wait(30)
                i += 1
                if i > 3:
                    self.room_detail_queue_task.put((False, community_building_model))
                    return False, community_building_model
            except Exception as e:
                my_logger.error(f"data_table: {data_table} : {e}, link: {lnk}")
                self.room_detail_queue_task.put((False, community_building_model))
                return False, community_building_model

    # 爬项目
    def crawler_project(self):
        def crawler(k, v):
            link = JinanHandler.community_url.format(1, k, self.crawler_args.project_type.value, '', '')
            with self.fetch_page(link) as page:
                all_counts, all_page = JinanHandler.get_all_page_count(page)
                if all_counts == 0:
                    my_logger.warning(f"没有数据：{link}")
                    return False, link
            self.collect_community_project(all_page, k, v)
        product_futures = [JinanHandler.community_executor.submit(crawler, k, v) for k, v in JinanHandler.zones_dict.items()]
        # wait(product_futures)
        JinanHandler.process_produce_futures(product_futures)
        # consumer_futures = []
        # while not self.community_queue_task.empty():
        #     try:
        #         args = self.community_queue_task.get_nowait()
        #         consumer_futures.append(JinanHandler.community_executor.submit(self.get_community_link, *args))
        #     except queue.Empty:  # 如果队列为空，则捕获异常并跳出循环
        #         break
        # success_task, fail_task = self.process_futures(consumer_futures)
        success_task, fail_task = JinanHandler.process_tasks(self.community_queue_task, JinanHandler.community_executor, self.get_community_link)
        if fail_task:
           my_logger.error(f"获取楼盘链接失败:{fail_task}")
        my_logger.debug(f"获取到项目数量:{len(success_task)}")
        # my_logger.debug(success_task)
        if success_task:
            JinanHandler.bulk_save_or_update(JiNanDataCollectionCommunity, success_task, [CrawlerId.COMMUNITY_ID])

    async def async_crawler_project(self):
        async def crawler(k, v):
            link = JinanHandler.community_url.format(1, k, self.crawler_args.project_type.value, '', '')
            async with self.async_fetch_page(link) as page:
                all_counts, all_page = await asyncio.to_thread(JinanHandler.get_all_page_count, page)
                if all_counts == 0:
                    return [], []
                return await self.async_collect_community_project(all_counts, k, v)
        futures = [JinanHandler.community_executor.submit(asyncio.run, crawler(k, v)) for k, v in JinanHandler.zones_dict.items()]
        success_result = []
        fail_result = []
        for idx, future in enumerate(as_completed(futures), start=1):
            try:
                # is_success, result = future.result()
                # if is_success:
                #     success_result.extend(result)
                # else:
                #     my_logger.warning(f"任务失败: {result}")
                #     fail_result.append(result)
                success_task, fail_task = future.result()
                # my_logger.debug(success_task)
                success_result.extend(success_task)
                fail_result.extend(fail_task)
                # my_logger.debug(f"任务执行进度: {idx} / {task_cnt}")
            except Exception as e:
                my_logger.error(f"Error execute task: {e}")
        if fail_result:
           my_logger.error(f"获取楼盘链接失败:{fail_result}")
        my_logger.debug(f"获取到项目数量:{len(success_result)}")
        if success_result:
            await JinanHandler.async_bulk_save_or_update(JiNanDataCollectionCommunity, success_result, [CrawlerId.COMMUNITY_ID])

    # 爬楼栋
    def crawler_building(self):
        with get_session() as session:
            statement = select(JiNanDataCollectionCommunity.collect_community_id)
            community_id_generator = session.exec(statement).yield_per(1000)
            # 楼栋
            my_logger.debug("开始爬取楼栋信息...")
            product_futures = [JinanHandler.building_executor.submit(self.get_community_building_info, community_id)
                       for community_id in community_id_generator]
        wait(product_futures)
        consumer_futures = []
        while not self.building_queue_task.empty():
            try:
                args = self.building_queue_task.get_nowait()
                consumer_futures.append(JinanHandler.building_executor.submit(self.get_community_building_link, *args))
            except queue.Empty:  # 如果队列为空，则捕获异常并跳出循环
                break
        need_crawler_community_building_model_list, fail_task = JinanHandler.process_futures(consumer_futures)
        JinanHandler.bulk_save_or_update(JiNanDataCollectionBuilding, need_crawler_community_building_model_list, [CrawlerId.BUILDING_ID])
        if fail_task:
            my_logger.warning(f"失败获取楼栋任务数量: {len(fail_task)}")
            my_logger.warning(f"失败获取楼栋任务: {fail_task}")
            # 失败的 building_id = None
            # JinanHandler.bulk_insert(CommunityBuildingRecord, fail_task)
        return need_crawler_community_building_model_list
    async def async_crawler_building(self):
        async for session in get_db():
            statement = select(JiNanDataCollectionCommunity.collect_community_id)
            community_project_link_generator = await session.exec(statement)
            community_project_link_generator = community_project_link_generator.yield_per(1000)
            # 楼栋
            my_logger.debug("开始爬取楼栋信息...")
            # futures = [building_executor.submit(self.async_get_community_building_info, community_model)
            #            for community_model in community_project_link_generator]
            futures = [self.async_get_community_building_info(community_id)
                       for community_id in community_project_link_generator]
        success_result, fail_result = await self.process_async_futures(futures)
        if fail_result:
            my_logger.warning(f"失败获取楼栋任务数量: {len(fail_result)}")
            my_logger.warning(f"失败获取楼栋任务: {fail_result}")
        return success_result
    # 爬某一个小区的楼栋
    def crawler_building_by_community_id(self, community_id):
        # 楼栋
        my_logger.debug("开始爬取楼栋信息...")
        self.get_community_building_info(community_id)
        consumer_futures = []
        while not self.building_queue_task.empty():
            try:
                args = self.building_queue_task.get_nowait()
                consumer_futures.append(JinanHandler.building_executor.submit(self.get_building_info, *args))
            except queue.Empty:  # 如果队列为空，则捕获异常并跳出循环
                break
        success_task, fail_task = self.process_futures(consumer_futures)
    # 爬房间
    def crawler_room(self):
        with get_session() as session:
            statement = select(CommunityBuildingRecord).where(CommunityBuildingRecord.is_house == '0')
            building_model_generator = session.exec(statement).yield_per(1000)
        futures = [JinanHandler.room_executor.submit(self.collect_community_table_and_room_info, building_model) for building_model in building_model_generator]
        my_logger.debug(f"爬取房间信息任务数量: {len(futures)}")
        success_task, fail_task = JinanHandler.process_futures(futures)
        # todo 失败任务处理
        my_logger.warning(f"失败获取房间任务数量: {len(fail_task)}")
        if fail_task:
            my_logger.warning(f"失败获取房间任务: {fail_task}")

    def need_crawler_room(self, building_model_list: list[JiNanDataCollectionBuilding]):
        product_futures = [JinanHandler.room_executor.submit(self.collect_community_table_and_room_info, building_model) for building_model in building_model_list]
        task_cnt = len(product_futures)
        my_logger.debug(f"爬取房间信息任务数量: {task_cnt}")
        # wait(product_futures)
        consumer_futures = []
        i = 1
        while i <= task_cnt:
            try:
                args = self.room_detail_queue_task.get()
                is_success, res = args
                if is_success:
                    consumer_futures.extend([JinanHandler.data_executor.submit(self.collect_house_info, model) for model in res])
                    if len(consumer_futures) >= 200:
                        success_task, fail_task = JinanHandler.process_futures(consumer_futures)
                        JinanHandler.bulk_save_or_update(JiNanDataCollectionRoom, success_task, [CrawlerId.ROOM_ID])
                        consumer_futures.clear()
                my_logger.debug(f"任务消费进度: {i} / {task_cnt}")
                i += 1
            except Exception as e:
                my_logger.error(f"Error execute task: {e}")
                continue
        success_task, fail_task = JinanHandler.process_futures(consumer_futures)
        JinanHandler.bulk_save_or_update(JiNanDataCollectionRoom, success_task, [CrawlerId.ROOM_ID])
        # todo 失败任务处理
        my_logger.warning(f"失败获取房间任务数量: {len(fail_task)}")
        if fail_task:
            my_logger.warning(f"失败获取房间任务: {fail_task}")

    # 爬房间详情
    def crawler_room_detail(self):
        with get_session() as session:
            stat = select(CommunityDataCollection.building_id).where(
                CommunityDataCollection.building_id != None, CommunityDataCollection.purpose == None).group_by(CommunityDataCollection.building_id)
            building_id_list = session.exec(stat).all()
        update_list = []
        update_futures = []
        for building_id in building_id_list:
            stat = select(CommunityDataCollection).where(CommunityDataCollection.building_id == building_id, CommunityDataCollection.purpose == None)
            building_data_generator = session.exec(stat).yield_per(1000)
            futures = []
            for building_data in building_data_generator:
                futures.append(JinanHandler.data_executor.submit(self.collect_house_info, building_data))
            success_task, fail_task = JinanHandler.process_futures(futures)
            update_list.extend(success_task)
            if len(update_list) >= 1000:
                # JinanHandler.bulk_update(CommunityDataCollection, update_list)
                update_futures.append(JinanHandler.executor.submit(self.update, deepcopy(update_list)))
                update_list.clear()
            if fail_task:
                my_logger.warning(f"失败获取房间详情, 数量{len(fail_task)}, 任务: {fail_task}")
        wait(update_futures)
        if update_list:
            JinanHandler.bulk_update(CommunityDataCollection, update_list)

    def crawler_by_name(self, vo: CrawlerCommunityByName):
        # county_id = JinanHandler.zones_dict.get(vo.county)
        county_id = 370102
        link = JinanHandler.community_url.format(1, county_id, self.crawler_args.project_type.value, vo.community_name, '')
        with self.fetch_page(link) as page:
            all_counts, all_page = JinanHandler.get_all_page_count(page)
            if all_counts == 0:
                my_logger.warning(f"没有找到小区: {vo.community_name}")
                return False, link
        self.collect_community_project(all_page, county_id, vo.county)
        self.crawler_args.project_name = vo.community_name
        consumer_futures = []
        while not self.community_queue_task.empty():
            try:
                args = self.community_queue_task.get_nowait()
                consumer_futures.append(JinanHandler.community_executor.submit(self.get_community_link, *args))
            except queue.Empty:  # 如果队列为空，则捕获异常并跳出循环
                break
        success_task, fail_task = self.process_futures(consumer_futures)
        my_logger.debug(success_task)
        product_futures = [JinanHandler.building_executor.submit(self.get_community_building_info, model.collect_community_id)
                           for model in success_task]
        wait(product_futures)
        consumer_futures.clear()
        while not self.building_queue_task.empty():
            try:
                args = self.building_queue_task.get_nowait()
                consumer_futures.append(JinanHandler.building_executor.submit(self.get_community_building_link, *args))
            except queue.Empty:  # 如果队列为空，则捕获异常并跳出循环
                break
        need_crawler_community_building_model_list, fail_task = JinanHandler.process_futures(consumer_futures)
        self.need_crawler_room(need_crawler_community_building_model_list)
        # for item in success_task:
        #     d = {}
        #     with get_session() as session:
        #         stat = select(JiNanDataCollectionBuilding).where(JiNanDataCollectionBuilding.community_id == item.collect_community_id)
        #         building_list = session.exec(stat).all()
        #         d['buildingList'] = building_list
        #         for building_model in building_list:


        # if success_task:
        #     JinanHandler.bulk_save_or_update(JiNanDataCollectionCommunity, success_task, [CrawlerId.COMMUNITY_ID])
        # self.crawler_building_by_community_id()

    # 更新单个楼栋信息
    def update_building(self, building_id):
        with get_session() as session:
            statement = select(CommunityBuildingRecord).where(CommunityBuildingRecord.building_id == building_id)
            community_building_model = session.exec(statement).one()
            # self.collect_community_building(community_building_model)
            session.add(community_building_model)
            session.commit()

    # 更新项目里所有的楼栋
    def update_community_building(self, community_id):
        with get_session() as session:
            statement = select(CommunityProjectRecord).where(CommunityProjectRecord.community_id == community_id)
            community_project_model = session.exec(statement).one()
        futures = [JinanHandler.executor.submit(self.get_community_building_info, community_project_model)]
        success_task, fail_task = JinanHandler.process_futures(futures)
        JinanHandler.bulk_insert(CommunityBuildingRecord, success_task)
        if fail_task:
            my_logger.warning(f"失败获取楼栋任务数量: {len(fail_task)}")
            my_logger.warning(f"失败获取楼栋任务: {fail_task}")

    # 更新楼栋的房间, 失败就bu_id重跑一遍
    def update_building_room(self, building_id):
        futures = []
        with get_session() as session:
            statement = select(CommunityBuildingRecord).where(CommunityBuildingRecord.building_id == building_id)
            building_data = session.exec(statement).one()
            stat = select(func.count()).select_from(CommunityDataCollection).where(CommunityDataCollection.building_id == building_id)
            count = session.exec(stat).one()
        my_logger.info(f"已爬该楼栋房间数: {count}")
        # 房间数小于售房数+可售房数，则爬取该楼栋数据
        if count < building_data.sale_count + building_data.can_sale_count:
            futures.append(JinanHandler.executor.submit(self.collect_community_table_and_room_info, building_data))
        else:
            my_logger.info(f"房间数无变化")
            return
        success_task, fail_task = JinanHandler.process_futures(futures)
        JinanHandler.bulk_insert(CommunityDataCollection, success_task)
        # 去重
        with (session):
            sql = """
                CREATE TEMPORARY TABLE temp_keep_ids AS
                SELECT id
                FROM (
                         SELECT id,
                                ROW_NUMBER() OVER(PARTITION BY building_id ORDER BY update_time DESC, is_house, id DESC) as row_num
                         FROM community_building_record
                     ) t
                WHERE t.row_num = 1;
                
                DELETE FROM community_building_record
                WHERE id NOT IN (SELECT id FROM temp_keep_ids);
        
                DROP TEMPORARY TABLE temp_keep_ids;
            """
            session.exec(sql)
            session.commit()
        if fail_task:
            my_logger.warning(f"失败获取room任务数量: {len(fail_task)}")
            # 失败的 room_id = None
            # JinanHandler.bulk_insert(CommunityDataCollection, fail_task)
            my_logger.warning(f"失败获取room任务: {fail_task}")

    # 更新某个房间的具体信息
    def update_room_info(self, room_id):
        with get_session() as session:
            stat = select(CommunityDataCollection).where(CommunityDataCollection.room_id == room_id)
            room_model = session.exec(stat).one()
            futures = [JinanHandler.data_executor.submit(self.collect_house_info, room_model)]
            wait(futures)
            session.add(room_model)
            session.commit()

    def crawler_start(self):
        # self.crawler_project()
        # need_crawler_building_list = self.crawler_building()
        # my_logger.info(f"需要爬取的楼栋数量{len(need_crawler_building_list)}")
        with get_session() as session:
            stat = select(JiNanDataCollectionBuilding)
            need_crawler_building_list = session.exec(stat).all()
        self.need_crawler_room(need_crawler_building_list)
    async def async_crawler_start(self):
        # 爬项目时, 判断未签套数, 变化就更新, 新增也更新
        # need_crawler_project_list = self.crawler_project()
        # my_logger.debug(f"需要爬取的项目数量: {len(need_crawler_project_list)}")
        # futures = []
        # for project in need_crawler_project_list:
        #     futures.append(executor.submit(self.crawler_building_by_community, project))
        # success, fail = JinanHandler.process_futures(futures)
        # await self.async_crawler_project()
        need_crawler_building_list = await self.async_crawler_building()
        my_logger.info(f"需要爬取的楼栋数量{len(need_crawler_building_list)}")

    @staticmethod
    def generate_unit_id(model: CommunityDataCollection):
        return model.community_id + model.building_id

    @staticmethod
    def convert_unit(unit_str: str):
        if not unit_str:
            return None
        unit = unit_str.replace('单元', '')
        num = change_to_int(unit)
        if num is None:
            my_logger.error(f"单元转换失败: {unit_str}")
            raise ValueError(f"单元转换失败: {unit_str}")
        return unit

    @staticmethod
    def convert_room(room_str: str):
        if not room_str:
            return "01"
        num = change_to_int(room_str)
        if num is None:
            my_logger.error(f"房间转换失败: {room_str}")
            raise ValueError(f"房间转换失败: {room_str}")
        if num < 10:
            return '0' + str(num)
        else:
            return str(num)

    @staticmethod
    def convert_room_number(room_number_str: str):
        if not room_number_str :
            return 1
        return change_to_int(room_number_str)

    @staticmethod
    def convert_floor(floor_str: str):
        if not floor_str:
            return None
        floor = floor_str.replace('F', '')
        num = change_to_int(floor)
        if num is None:
            my_logger.error(f"楼层转换失败: {floor_str}")
            raise ValueError(f"楼层转换失败: {floor_str}")
        return num

    @staticmethod
    def convert_total_unit(unit_str: str):
        if not unit_str:
            return 1
        num = change_to_int(unit_str)
        if num is None:
            my_logger.error(f"单元转换失败: {unit_str}")
            raise ValueError(f"单元转换失败: {unit_str}")
        return num

    @staticmethod
    def get_community_link_id(community_link: str):
        query_params = get_query_params(community_link)
        # 提取 prjno 参数
        community_id = query_params.get(JiNanUrlId.PROJECT, None)[0]
        # print(f'提取的楼栋id: {community_id}')
        return community_id
    @staticmethod
    def get_community_building_link_id(community_building_link: str):
        query_params = get_query_params(community_building_link)
        community_id = query_params.get(JiNanUrlId.BUILDING, None)[0]
        return community_id
    @staticmethod
    def get_community_table_id(community_table_link: str):
        query_params = get_query_params(community_table_link)
        community_id = query_params.get(JiNanUrlId.TABLE, None)[0]
        return community_id

    @staticmethod
    def get_all_page_count(page):
        try:
            all_counts = page.ele('#allcounts').attr('value')
            all_page = page.ele('#allpage').attr('value')
            my_logger.debug(f"数据总数量:{all_counts}, 总页数:{all_page}")
        except (ElementNotFoundError, WaitTimeoutError) as e:
            my_logger.warning(f"Error get_all_page_count task: {e}")
            return 0, 0
        return int(all_counts), int(all_page)
    def test(self):
        with get_session() as session:
            stat = select(func.count()).select_from(CommunityProjectRecord)
            count = session.exec(stat).one()
            print(count)

if __name__ == '__main__':
    # co = ChromiumOptions().headless()
    co = ChromiumOptions()
    co.set_retry(2, 30)
    # co.set_load_mode('eager')
    co.set_user_agent(UserAgent().random)
    co.incognito()
    co.ignore_certificate_errors()
    b = Chromium(co)
    # asyncio.run(JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).async_crawler_start())
    JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).crawler_start()
    t = CrawlerCommunityByName(community_name='绿地', province='山东', area='济南', county='历下区')
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).crawler_by_name(t)
    # b.quit()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).test()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).need_crawler_room([])
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).update_building_room('11a6b47f-aa56-477c-871a-a318b792a500')
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).crawler_project()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).crawler_building()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).crawler_room()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).crawler_room_detail()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).get_house_info()
    # JinanHandler(b, CrawlerJiNanArgs(project_type=ProjectType.HOUSE)).process_community_table(JinanHandler.build_table_url_prefix + '683392e9-ee07-48b7-acd2-fc390bc4c4f2')